package com.xjscrm.console.chat.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.uzai.mobile.collect.api.dto.trigger.talk.MicaTalkContentTextMsg;
import com.uzai.trace.TraceGenerate;
import com.xjscrm.console.chat.cache.CacheHolder;
import com.xjscrm.console.chat.dispatch.match.AllocateRule;
import com.xjscrm.console.chat.dispatch.match.DispatchCondition;
import com.xjscrm.console.chat.enums.ChatTalkRecordSource;
import com.xjscrm.console.chat.enums.MQMsgType;
import com.xjscrm.console.chat.frame.dto.ChatRecordDTO;
import com.xjscrm.console.chat.frame.dto.ChatSessionDTO;
import com.xjscrm.console.chat.mq.MQSender;
import com.xjscrm.console.chat.mq.dto.DispatchDTO;
import com.xjscrm.console.chat.mq.dto.SendDTO;
import com.xjscrm.console.chat.mq.dto.SendMQDTO;
import com.xjscrm.console.chat.utils.MessageUtil;
import com.xjscrm.console.dto.chat.setting.ChatAutoReplyJsonDTO;
import com.xjscrm.console.entity.ChatSettings;
import com.xjscrm.console.entity.ChatTalkRecord;
import com.xjscrm.console.entity.ChatTalkSession;
import com.xjscrm.console.entity.ChatTalkSessionStaffRel;
import com.xjscrm.console.mq.producer.SendMsgProducer;
import com.xjscrm.console.service.customer.chat.*;
import com.xjscrm.console.vo.chat.ChatDispatchSettingsVO;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author mingyang.chen create in 2024/3/12 14:10
 */
@RocketMQMessageListener(topic = "XJ_WW_DISPATCH_TOPIC", consumerGroup = "XJ_WW_DISPATCH_GROUP", messageModel = MessageModel.CLUSTERING)
@Component
@Slf4j
public class MsgDispatchConsumer implements RocketMQListener<DispatchDTO> {

    @Autowired
    private ChatSettingsService chatSettingsService;
    @Autowired
    private Environment environment;
    @Autowired
    private CacheHolder cacheHolder;
    @Autowired
    private ChatDispatchSettingsService dispatchSettingsService;
    @Autowired
    private ChatTalkSessionService chatTalkSessionService;
    @Autowired
    private ChatTalkRecordService chatTalkRecordService;
    @Autowired
    private ChatTalkSessionStaffRelService chatTalkSessionStaffRelService;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private SendMsgProducer sendMsgProducer;

    // 规则
    @Autowired
    private DispatchCondition keywordsDispatchCondition;
    @Autowired
    private AllocateRule staffAllocateRule;

    @Override
    @TraceGenerate
    public void onMessage(DispatchDTO message) {

        try {
            String mqTag = environment.getProperty("mqTag");

            log.info("分发聊天消息: 收到消息: {}", JSON.toJSONString(message));
            Long merId = message.getMerId();
            Long deviceUniqueId = message.getDeviceUniqueId();
            Long talkSessionId = message.getTalkSessionId();
            //
            ChatSettings chatSettings = chatSettingsService.selectByMerId(merId);
            ChatDispatchSettingsVO dispatchSettings = dispatchSettingsService.get(merId);
            //
            ChatTalkSession talkSession = chatTalkSessionService.getById(merId, talkSessionId);
            if (Objects.isNull(talkSession)) {
                log.error("分发聊天消息: 会话未找到");
                return;
            }
            Long talkRecordId = talkSession.getLastRecvId();
            ChatTalkRecord talkRecord = chatTalkRecordService.getById(merId, talkRecordId);
            if (Objects.isNull(talkRecord)) {
                log.error("分发聊天消息: 会话的聊天记录未找到");
            }
            //
            if (Objects.equals(talkSession.getOpenShield(), 1)) {
                log.error("分发聊天消息: 会话已被屏蔽, 消息分发失败.");
                return;
            }
            //
            if (!matchShieldKeyWord(talkRecord, chatSettings)) {
                return;
            }
            //
            if (Objects.equals(talkRecord.getContentType(), 1)) {
                MicaTalkContentTextMsg textMsg = JSON.parseObject(talkRecord.getContent(), MicaTalkContentTextMsg.class);
                String text = textMsg.getText();
                Boolean match = keywordsDispatchCondition.match(text.trim(), dispatchSettings.getDispatchConditionConf());
                if (!match) {
                    log.error("分发聊天消息: 分发失败, 未通过分发关键词过滤: {}", text);
                    return;
                }
            }

            String staffTag;
            Long staffId;
            //
            ChatTalkSessionStaffRel sessionStaffRel = chatTalkSessionStaffRelService.getByTalkId(merId, deviceUniqueId, talkSessionId);
            if (Objects.isNull(sessionStaffRel)) {
                //
                staffId = staffAllocateRule.allocateStaffId(merId, deviceUniqueId, dispatchSettings.getAllocateRuleConf());
                if (Objects.isNull(staffId)) {
                    log.error("分发聊天消息: 当前聊天消息的机器人为分配员工, 停止分发");
                    return;
                }
                // 排队
                boolean isQueue = Objects.equals(staffId, -100L);
                boolean isLeave = Objects.equals(staffId, -101L);
                // 判断是否开启排队
                if (isLeave || isQueue) {
                    //
                    if (Objects.isNull(dispatchSettings.getSessionAccessConf()) || Objects.equals(1, ObjectUtils.defaultIfNull(dispatchSettings.getSessionAccessConf().getOpenOrder(), 1))) {
                        if (Objects.equals(talkRecord.getSource(), ChatTalkRecordSource.receive.getCode()) && !cacheHolder.containsSessionOrderRepeatTime(deviceUniqueId, talkSessionId)) {
                            this.sendOrderReply(talkSession, dispatchSettings.getAutoReplyConf());
                        }
                        if (isQueue) {
                            List<Long> longs = cacheHolder.allSessionOrder(merId, deviceUniqueId);
                            if (longs.size() >= 300) {
                                cacheHolder.leftPopSessionOrder(merId, deviceUniqueId);
                            }
                            // 入队等待
                            cacheHolder.setSessionOrder(merId, deviceUniqueId, talkSessionId);
                            cacheHolder.delSessionLeave(merId, deviceUniqueId, talkSessionId);
                            log.error("分发聊天消息: 当前机器人-[{}]未关联员工/员工都未上线, 进入排队", deviceUniqueId);
                            return;
                        }
                        List<Long> longs = cacheHolder.allSessionLeave(merId, deviceUniqueId);
                        if (longs.size() >= 300) {
                            cacheHolder.leftPopSessionLeave(merId, deviceUniqueId);
                        }
                        // 入队等待
                        cacheHolder.setSessionLeave(merId, deviceUniqueId, talkSessionId);
                        cacheHolder.delSessionOrder(merId, deviceUniqueId, talkSessionId);
                        log.error("分发聊天消息: 当前机器人-[{}]未关联员工/员工都未上线, 进入留言", deviceUniqueId);
                        return;
                    }
                    //
                    else {
                        log.error("分发聊天消息: 当前运营商未开启排队留言开关");
                        return;
                    }
                }
                log.info("分发聊天消息: 会话分配的员工: {}, 会话id: {}", staffId, talkSessionId);
                //
                chatTalkSessionStaffRelService.addIfAbsent(merId, talkSession, staffId);
                staffTag = cacheHolder.getStaffTag(staffId);
            } else {

                staffId = sessionStaffRel.getStaffId();
                staffTag = cacheHolder.getStaffTag(staffId);
                //
                if (StringUtils.isBlank(staffTag)) {
                    log.error("分发聊天消息: 会话: {} 分配的员工: {} 未上线, 删除关联记录, 重新分发", talkSessionId, staffId);
                    //
                    chatTalkSessionStaffRelService.clean(merId, sessionStaffRel.getId());
                    //
                    rocketMQTemplate.asyncSend("XJ_WW_DISPATCH_TOPIC", JSON.toJSONString(message), new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            log.info("分发聊天消息: 重新分发消息成功 ->  msg:{}", JSON.toJSONString(message));
                        }

                        @Override
                        public void onException(Throwable throwable) {
                            log.error("分发聊天消息: 重新分发消息失败", throwable);
                        }
                    });
                    return;
                }
                //
                chatTalkSessionStaffRelService.cleanEnd(merId, sessionStaffRel.getId());
                //
                chatTalkSessionService.delChatSendDelayTask(merId, sessionStaffRel.getStaffId(), talkSession);
                chatTalkSessionService.delChatRecvDelayTask(merId, sessionStaffRel.getStaffId(), talkSession);
                chatTalkSessionService.delChatCleanDelayTask(merId, sessionStaffRel.getStaffId(), talkSession);
                //
            }

            //
            if (StringUtils.equals(staffTag, mqTag)) {
                ChatSessionDTO chatSessionDTO = new ChatSessionDTO();
                BeanUtils.copyProperties(talkSession, chatSessionDTO);
                ChatRecordDTO chatRecordDTO = new ChatRecordDTO();
                BeanUtils.copyProperties(talkRecord, chatRecordDTO);
                chatRecordDTO.parse();

                if (Objects.nonNull(sessionStaffRel)) {
                    chatSessionDTO.setHang(sessionStaffRel.getHang());
                    chatSessionDTO.setTransfer(sessionStaffRel.getTransfer());
                    chatSessionDTO.setAuto(sessionStaffRel.getAuto());
                }
                Channel staffChannel = cacheHolder.getStaffChannel(staffId);
                cacheHolder.delSessionOrder(merId, talkSession.getDeviceUniqueId(), talkSessionId);
                cacheHolder.delSessionOrderRepeatTime(talkSession.getDeviceUniqueId(), talkSessionId);
                chatTalkSessionService.addChatSendDelayTask(merId, staffId, talkSession);

                MessageUtil.sendNewSessionTalk(staffChannel, chatSessionDTO, chatRecordDTO);
            } else {

                SendDTO sendDTO = new SendDTO();
                sendDTO.setStaffId(staffId);
                sendDTO.setTalkSessionId(talkSessionId);
                sendDTO.setTalkRecordId(talkRecordId);
                sendDTO.setMerId(merId);
                staffTag = cacheHolder.getStaffTag(staffId);
                sendDTO.setStaffTag(staffTag);

                SendMQDTO sendMQDTO = new SendMQDTO();
                sendMQDTO.setType(MQMsgType.push_talk.getCode());
                sendMQDTO.setData(sendDTO);

                log.info("员工未在当前节点, 路由到目标节点: {}", JSON.toJSONString(sendDTO));
                MQSender.sendMQMsg("XJ_WW_DISPATCH_HANDLE_TOPIC:" + staffTag, sendMQDTO);
            }
        } catch (Exception e) {
            log.error("分发聊天消息: 处理异常", e);
        }
    }

    private boolean matchShieldKeyWord(ChatTalkRecord talkRecord, ChatSettings chatSettings) {
        if (Objects.isNull(talkRecord)) {
            return true;
        }
        if (Objects.equals(talkRecord.getContentType(), 1)) {
            MicaTalkContentTextMsg textMsg = JSON.parseObject(talkRecord.getContent(), MicaTalkContentTextMsg.class);
            String text = textMsg.getText();

            if (Objects.nonNull(chatSettings) && Objects.equals(chatSettings.getOpenKeywordShield(), 1)) {

                String blurKeyword = chatSettings.getBlurKeyword();
                String exactKeyword = chatSettings.getExactKeyword();

                if (StringUtils.isNotEmpty(blurKeyword)) {
                    List<String> blurKeywords = JSON.parseArray(blurKeyword, String.class);
                    if (blurKeywords.stream().anyMatch(item -> StringUtils.contains(text, item))) {
                        log.error("分发聊天消息: 分发失败, 匹配到了屏蔽模糊关键词");
                        return false;
                    }
                }
                if (StringUtils.isNotEmpty(exactKeyword)) {
                    List<String> exactKeywords = JSON.parseArray(exactKeyword, String.class);
                    if (exactKeywords.stream().anyMatch(item -> StringUtils.equals(text, item))) {
                        log.error("分发聊天消息: 分发失败, 匹配到了屏蔽精准关键词");
                        return false;
                    }
                }
            }
        }
        return true;
    }

    private void sendOrderReply(ChatTalkSession talkSession, ChatAutoReplyJsonDTO autoReplyJson) {
        if (Objects.isNull(talkSession) || Objects.equals(talkSession.getType(), 111)) {
            return;
        }

        if (!Objects.equals(autoReplyJson.getOpenQueueReply(), 1)) {
            return;
        }
        boolean b = cacheHolder.containsSessionOrderRepeatTime(talkSession.getDeviceUniqueId(), talkSession.getId());
        if (b) {
            log.error("分发聊天消息: 此回话已回复过排队消息");
            return;
        }
        String r = autoReplyJson.getQueueReplyText();
        Integer queueRepeatTime = autoReplyJson.getQueueRepeatTime();
        if (StringUtils.isBlank(r) || Objects.isNull(queueRepeatTime) || queueRepeatTime <= 0) {
            log.error("分发聊天消息: 排队回复配置有误, 文本: {}, 时间: {}", r, queueRepeatTime);
            return;
        }
        ChatTalkRecord chatTalkRecord = chatTalkRecordService.buildTextRecord(talkSession.getMerId(), -1L, talkSession, r);
        sendMsgProducer.replySessionText(talkSession, chatTalkRecord);
        cacheHolder.setSessionOrderRepeatTime(talkSession.getDeviceUniqueId(), talkSession.getId(), queueRepeatTime.longValue(), TimeUnit.MINUTES);
    }
}
